package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: gaojian
 * @Description: 针对企查查t_eci_operateexception表和中数t_ent_ope_abnormity表进行数据映射开发
 * @Date: 2021/12/29
 */

public class TEciOperateexception {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_eci_operateexception").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_ent_ope_abnormity");
        long zs_count = zs_data.count();
        if(zs_count>0){

            // 处理步骤说明 1.UPDATE 2.INSERT 3.DELETE
            // 1.UPDATE
            // SQL逻辑：定义更新集合 update_data
            //          1.从 dc_ep_ods 取得中数表全集 统一社会信用代码 不为空 为集合A
            //          2.从 xy_ods 取得企查查全集 统一社会信用代码 不为空 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做INNER JOIN，注意要 UPPER 和 TRIM 函数 为集合C
            //              key_no，company_id 来自企查查
            //              JOB_ID, "0" as isadd 两个字段是要增加的
            //          4.从 dc_ep_ods 取得中数表全集 组织机构号 不为空 并且 统一社会信用代码为空的 为集合D
            //          5.从 xy_ods 取得企查查全集 组织机构号 不为空 为集合E
            //          6.集合D 和 集合E 通过组织机构号，做INNER JOIN 注意要 UPPER 和 TRIM 函数 为集合F
            //              key_no，company_id 来自企查查
            //              JOB_ID, "0" as isadd 两个字段是要增加的
            //          7.集合C 与 机会F 使用 UNION 为并集 G
            //          8.update_data.createOrReplaceTempView 更新
            //          9.同步到目标表

            Dataset<Row> update_data = spark.sql("SELECT \n" +
                    "B.ID,\n" +
                    "B.KEY_NO,\n" +
                    "B.COMPANY_ID,\n" +
                    "B.COMPANY_NAME,\n" +
                    "A.INDATE AS ADD_DATE,\n" +
                    "A.INREASON AS ADD_REASON,\n" +
                    "A.YR_REGORG AS DECISION_OFFICE,\n" +
                    "A.OUTDATE AS REMOVE_DATE,\n" +
                    "A.OUTREASON AS REMOVE_REASON,\n" +
                    "A.YC_REGORG AS REMOVE_DECISION_OFFICE,\n" +
                    "A.JOBID, '0' AS ISADD \n" +
                    "FROM \n" +
                    "(\n" +
                    "    select a.* from a_dc_ep_ods.zs_t_ent_ope_abnormity a \n" +
                    "    where a.creditcode is not null and a.creditcode != '' \n" +
                    ") A INNER JOIN \n" +
                    "(\n" +
                    "    select b.credit_code,b.company_name,a.id,a.key_no,a.company_id,a.add_date,a.add_reason,a.decision_office,a.remove_date,a.romove_reason,a.remove_decision_office,a.dates,a.isadd from a_dc_ep_dwi.t_eci_operateexception a \n" +
                    "    inner join a_dc_ep_ods.t_eci_company b on b.company_id = a.company_id \n" +
                    "    where b.credit_code is not null and b.credit_code!='' \n" +
                    ") B \n" +
                    "ON UPPER(TRIM(B.CREDIT_CODE)) = UPPER(TRIM(A.CREDITCODE)) \n" +
                    "WHERE date_format(B.ADD_DATE,'yyyy-MM-dd') = date_format(A.indate,'yyyy-MM-dd')\n" +
                    "AND UPPER(TRIM(B.DECISION_OFFICE)) = UPPER(TRIM(A.YR_REGORG))")
                    .union(spark.sql("SELECT \n" +
                            "B.ID,\n" +
                            "B.KEY_NO,\n" +
                            "B.COMPANY_ID,\n" +
                            "B.COMPANY_NAME,\n" +
                            "A.INDATE AS ADD_DATE,\n" +
                            "A.INREASON AS ADD_REASON,\n" +
                            "A.YR_REGORG AS DECISION_OFFICE,\n" +
                            "A.OUTDATE AS REMOVE_DATE,\n" +
                            "A.OUTREASON AS REMOVE_REASON,\n" +
                            "A.YC_REGORG AS REMOVE_DECISION_OFFICE,\n" +
                            "A.JOBID, '0' AS ISADD \n" +
                            "FROM \n" +
                            "(\n" +
                            "    select a.* from a_dc_ep_ods.zs_t_ent_ope_abnormity a \n" +
                            "    where a.regno is not null and a.regno != '' \n" +
                            "    and (a.creditcode = '' or a.creditcode is null)\n" +
                            ") A INNER JOIN \n" +
                            "(\n" +
                            "    select a.id,a.key_no,a.company_id,a.add_date,a.add_reason,a.decision_office,a.remove_date,a.romove_reason,a.remove_decision_office,a.dates,a.isadd,b.company_name,b.no from a_dc_ep_dwi.t_eci_operateexception a \n" +
                            "    inner join a_dc_ep_ods.t_eci_company b on b.company_id = a.company_id \n" +
                            "    where b.no is not null and b.no != '' and (b.credit_code is null or b.credit_code = '')\n" +
                            ") B \n" +
                            "ON UPPER(TRIM(B.NO)) = UPPER(TRIM(A.REGNO)) \n" +
                            "WHERE date_format(B.ADD_DATE,'yyyy-MM-dd') = date_format(A.indate,'yyyy-MM-dd')\n" +
                            "AND UPPER(TRIM(B.DECISION_OFFICE)) = UPPER(TRIM(A.YR_REGORG))" ));
            update_data.createOrReplaceTempView("tmp_t_eci_operateexception");
            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_operateexception select * from tmp_t_eci_operateexception");

            // 2.INSERT
            // SQL逻辑：定义 insert_data 集合
            //          1.从 dc_ep_ods 取得中数表全集 统一社会信用代码 不为空 为集合A
            //          2.从 xy_ods 取得企查查全集 统一社会信用代码 不为空 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做LEFT JOIN，注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合C
            //                  产生 key_no 和 company_id 使用 md5(concat(集合A.entname,if(length(集合A.creditcode)=18,集合A.creditcode,'')))
            //                  JOB_ID, "1" as isadd 两个字段是要增加的
            //          4.从 dc_ep_ods 取得中数表全集 组织机构号 不为空 并且 统一社会信用代码为空的 为集合D
            //          5.从 xy_ods 取得企查查全集 组织机构号 不为空 为集合E
            //          6.集合D 和 集合E 通过组织机构号，做 LEFT JOIN 注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合F
            //                  产生 key_no 和 company_id 使用 md5(concat(集合A.entname))
            //                  JOB_ID, "1" as isadd 两个字段是要增加的
            //          7.集合C 与 机会F 使用 UNION 为并集 G
            //          8.insert_data.createOrReplaceTempView 完成插入
            //          9.同步到目标表
            Dataset<Row> insert_data = spark.sql("SELECT\n" +
                    "MD5(CONCAT(UPPER(TRIM(A.entname)),UPPER(TRIM(A.YR_REGORG)),DATE_FORMAT(IF(A.indate = '' OR A.indate IS NULL,'1900-01-01',A.indate),'yyyy-MM-dd'))) AS ID,\n" +
                    "if(c.credit_code is not null,c.company_id,md5(concat(A.entname,if(length(A.creditcode)=18,A.creditcode,'')))) AS KEY_NO, \n" +
                    "if(c.credit_code is not null,c.company_id,md5(concat(A.entname,if(length(A.creditcode)=18,A.creditcode,'')))) AS COMPANY_ID, \n" +
                    "A.entname AS COMPANY_NAME,\n" +
                    "A.INDATE AS ADD_DATE,\n" +
                    "A.INREASON AS ADD_REASON,\n" +
                    "A.YR_REGORG AS DECISION_OFFICE,\n" +
                    "A.OUTDATE AS REMOVE_DATE,\n" +
                    "A.OUTREASON AS REMOVE_REASON,\n" +
                    "A.YC_REGORG AS REMOVE_DECISION_OFFICE,\n" +
                    "A.JOBID, '1' AS ISADD\n" +
                    "FROM \n" +
                    "(\n" +
                    "    select a.* from a_dc_ep_ods.zs_t_ent_ope_abnormity a\n" +
                    "    where creditcode is not null and creditcode != '' \n" +
                    ") A \n" +
                    "LEFT JOIN \n" +
                    "(\n" +
                    "    select a.*,b.credit_code from a_dc_ep_dwi.t_eci_operateexception a \n" +
                    "    inner join a_dc_ep_ods.t_eci_company b on b.company_id = a.company_id \n" +
                    "    where b.credit_code is not null and b.credit_code != '' \n" +
                    ") B \n" +
                    "ON upper(trim(B.credit_code)) = upper(trim(A.creditcode))\n" +
                    "AND DATE_FORMAT(B.ADD_DATE,'yyyy-MM-dd') = DATE_FORMAT(A.indate,'yyyy-MM-dd')\n" +
                    "AND UPPER(TRIM(B.DECISION_OFFICE)) = UPPER(TRIM(A.YR_REGORG))\n " +
                     "left join a_dc_ep_ods.t_eci_company c on a.creditcode = c.credit_code " +
                    "WHERE B.CREDIT_CODE IS NULL")
                    .union(spark.sql("SELECT \n" +
                            "MD5(CONCAT(UPPER(TRIM(A.entname)),UPPER(TRIM(A.YR_REGORG)),DATE_FORMAT(IF(A.indate = '' OR A.indate IS NULL,'1900-01-01',A.indate),'yyyy-MM-dd'))) AS ID,\n" +
                            "if(c.COMPANY_NAME is not null,c.company_id,md5(concat(A.entname))) AS KEY_NO, \n" +
                            "if(c.COMPANY_NAME is not null,c.company_id,md5(concat(A.entname))) AS COMPANY_ID, \n" +
                            "A.entname AS COMPANY_NAME,\n" +
                            "A.INDATE AS ADD_DATE,\n" +
                            "A.INREASON AS ADD_REASON,\n" +
                            "A.YR_REGORG AS DECISION_OFFICE,\n" +
                            "A.OUTDATE AS REMOVE_DATE,\n" +
                            "A.OUTREASON AS REMOVE_REASON,\n" +
                            "A.YC_REGORG AS REMOVE_DECISION_OFFICE,\n" +
                            "A.JOBID, '1' AS ISADD\n" +
                            "FROM \n" +
                            "(\n" +
                            "    select a.* from a_dc_ep_ods.zs_t_ent_ope_abnormity a\n" +
                            "    where a.regno is not null and a.regno !='' and (a.creditcode is null or a.creditcode = '') \n" +
                            ") A\n" +
                            "LEFT JOIN\n" +
                            "(\n" +
                            "    select a.*,b.no as regno from a_dc_ep_dwi.t_eci_operateexception a \n" +
                            "    inner join a_dc_ep_ods.t_eci_company b on b.company_id = a.company_id \n" +
                            "    where b.no is not null and b.no != '' and (b.credit_code is null or b.credit_code = '') \n" +
                            ") B \n" +
                            "ON UPPER(TRIM(B.REGNO)) = UPPER(TRIM(A.REGNO))\n" +
                            "AND DATE_FORMAT(B.ADD_DATE,'yyyy-MM-dd') = DATE_FORMAT(A.indate,'yyyy-MM-dd')\n" +
                            "AND UPPER(TRIM(B.DECISION_OFFICE)) = UPPER(TRIM(A.YR_REGORG))\n" +
                            "left join a_dc_ep_ods.t_eci_company c on regexp_replace(trim(a.ENTNAME),'([(（）)])','') = regexp_replace(trim(C.COMPANY_NAME),'([(（）)])','') " +
                            "WHERE B.REGNO IS NULL"));
            insert_data.createOrReplaceTempView("tmp_t_eci_operateexception1");
            spark.sql("insert into table a_dc_ep_incr.t_eci_operateexception select * from tmp_t_eci_operateexception1");

            // 3.DELETE
            // SQL逻辑：定义 delete_data 集合
            //          1.从 xy_ods 取得企查查全集 为集合A
            //          2.从 dc_ep_ods 取得( 以_del结尾表为主表，left join原始表，找到删除的creditcode)  中数删除 全集 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做INNER JOIN，注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合C
            //                  取得企查查侧全部字段 并加
            //                  JOB_ID, "-1" as isadd 两个字段是要增加的
            //          4.delete_data.createOrReplaceTempView 完成插入
            //          9.同步到目标a_dc_ep_incr库表
            Dataset<Row> delete_data = spark.sql("SELECT \n" +
                    "B.ID,B.KEY_NO,B.COMPANY_ID,B.COMPANY_NAME,\n" +
                    "B.ADD_DATE,B.ADD_REASON,B.DECISION_OFFICE,\n" +
                    "B.REMOVE_DATE,B.ROMOVE_REASON,B.REMOVE_DECISION_OFFICE,\n" +
                    "A.JOBID,'-1' AS ISADD\n" +
                    "FROM\n" +
                    "(\n" +
                    "    select del.creditcode,del.jobid from a_dc_ep_ods.zs_t_ent_ope_abnormity_del del\n" +
                    "    left join a_dc_ep_ods.zs_t_ent_ope_abnormity a\n" +
                    "    on upper(trim(a.entid)) = upper(trim(del.entid))\n" +
                    "    where a.entid is null \n" +
                    ") A\n" +
                    "INNER JOIN\n" +
                    "(\n" +
                    "    select a.*,b.credit_code from a_dc_ep_dwi.t_eci_operateexception a \n" +
                    "    inner join a_dc_ep_ods.t_eci_company b on b.company_id = a.company_id \n" +
                    "    where b.credit_code is not null and b.credit_code != '' \n" +
                    ") B\n" +
                    "ON upper(trim(B.credit_code)) = upper(trim(A.creditcode))");
            delete_data.createOrReplaceTempView("tmp_t_eci_operateexception2");
            spark.sql("insert into table a_dc_ep_incr.t_eci_operateexception select * from tmp_t_eci_operateexception2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_eci_operateexception");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
